﻿//***********************************************************************
// Assembly         : IGolovkine.AoA.Tcp.Channels.Http
// Author           : Ilia
// Created          : 06-11-2011
//
// Last Modified By : Ilia
// Last Modified On : 01-10-2010
// Description      : 
//
// Copyright        : (c) Microsoft. All rights reserved.
//***********************************************************************
using System;
using System.Net;
using System.Net.Cache;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using IGolovkine.AoA.Diagnostics;

namespace IGolovkine.AoA.Tcp.Channels.Http
{
    public class HttpChannelClient : IMessageChannel, IDisposable
    {
        private readonly Subject<Message> _input;
        private readonly Subject<Message> _output;

        private readonly IDisposable _taskSubscription;
        private readonly IDisposable _inputSubscription;
        
        private readonly MessageAggregator _buffer;
        
        private long _connectionDataBytes;
        private long _allBytes;

        private readonly AutoResetEvent _messageEvent;
        private readonly HttpChannelClientConfiguration _settings;
        private readonly HttpEnvelopeExchangeStrategy _compressionStrategy;

        private readonly ILogger _logger;

        public HttpChannelClient(HttpChannelClientConfiguration settings)
        {
            ServicePointManager.Expect100Continue = false;

            _logger = ApplicationContext.LogManager.GetClassLogger(this);

            _input = new Subject<Message>();
            _output = new Subject<Message>();
            _settings = settings;
            
            _compressionStrategy = new HttpEnvelopeExchangeStrategy();

            _buffer = new MessageAggregator();
            _messageEvent = new AutoResetEvent(false);
            
            _inputSubscription = _input
                .Do((m) => this.WakeUpOnActivity())
                .Subscribe(_buffer.Input);

            bool isCancelled = false;

            Task task = Task.Factory.StartNew(() =>
            {
                try
                {
                    do
                    {
                        this.DoDataExchange();
                        _messageEvent.WaitOne(this.GetNextPollInterval());
                    }
                    while (isCancelled == false);

                }
                catch (Exception ex)
                {
                    _output.OnError(ex);
                }
                finally
                {
                    _output.OnCompleted();
                }
            });

            _taskSubscription = Disposable.Create(() =>
                {
                    isCancelled = true;
                    _messageEvent.Set();
                    task.Wait();
                });
        }

        private int _emptyPostCount/* = 0*/;

        private void DoDataExchange()
        {
            _logger.Trace("About to do data exchange...");
            HttpEnvelope outgoingMessage = new HttpEnvelope(_buffer.DequeueAll());

            if (outgoingMessage.GetMessages().Length > 0)
            {
                _logger.Debug("{0} messages dequeued", outgoingMessage.GetMessages().Length);
                foreach (Message message in outgoingMessage.GetMessages())
                {
                    _logger.Debug(message.ToString());
                }

            }
            _logger.Debug("Sending request...");
            HttpWebRequest request = this.CreateRequest();
            string compressionAlgoName = _compressionStrategy.Send(outgoingMessage, request);
            _logger.Debug("Request sent, length: {0}", request.ContentLength);

            _logger.Debug("Waiting for response...");
            WebResponse response = request.GetResponse();
            _logger.Debug("Got response, length: {0}", response.ContentLength);

            _allBytes += request.ContentLength + response.ContentLength;
            _connectionDataBytes += outgoingMessage.GetConnectionDataLength();

            HttpEnvelope incomingMessage = _compressionStrategy.Receive(response);
            _connectionDataBytes += incomingMessage.GetConnectionDataLength();

            foreach (Message message in incomingMessage.GetMessages())
            {
                _output.OnNext(message);
            }

            if (request.ContentLength > 0 || response.ContentLength > 0)
            {
                _emptyPostCount = 0;
                double overhead = ((_allBytes * 1.0 / _connectionDataBytes) - 1.0) * 100;
                _logger.Debug("Tx: {0:0000} Rx: {1:0000} OH: {2:00}% Comp = '{3}' Buffer = {4}", request.ContentLength, response.ContentLength, overhead, compressionAlgoName, _buffer.Count);
            }
            else
            {
                _emptyPostCount++;
            }
        }

        private void WakeUpOnActivity()
        {
            if (_isInIdleMode)
            {
                _logger.Debug("Waking up!");
                _messageEvent.Set();
            }
        }

        private HttpWebRequest CreateRequest()
        {
            HttpWebRequest request = (HttpWebRequest) HttpWebRequest.Create(_settings.ServerUrl);
            request.Method = "POST";
            request.ContentLength = 0;
            if (_settings.Proxy != null)
            {
                request.Proxy = _settings.Proxy;
            }
            request.CachePolicy = new RequestCachePolicy(RequestCacheLevel.BypassCache);
            return request;
        }

        private bool _isInIdleMode/* = false*/;

        private TimeSpan GetNextPollInterval()
        {
            double timeSinceLastActivity = (DateTime.Now - _buffer.LastMessage).TotalMilliseconds;
            
            double max = _settings.MaxPollInterval.TotalMilliseconds;
            double min = _settings.MinPollInterval.TotalMilliseconds;

            if (timeSinceLastActivity > _settings.IdleModeTrigger.TotalMilliseconds)
            {
                double nextPollInterval = Math.Min(Math.Max(min, timeSinceLastActivity), max);
                _isInIdleMode = true;
                _logger.Debug( "Entering idle mode, next poll will be in {0}ms", nextPollInterval);
                return TimeSpan.FromMilliseconds(nextPollInterval);
            }
            else
            {
                _isInIdleMode = false;
                return TimeSpan.FromMilliseconds(min);
            }
        }

        public IObservable<Message> Output
        {
            get { return _output; }
        }

        public IObserver<Message> Input
        {
            get { return _input; }
        }

        public void Dispose()
        {
            _logger.Debug("Diposed!");
            _inputSubscription.Dispose();
            _taskSubscription.Dispose();
        }
    }
}
